package com.raymond.cmpp.client.thread;

import com.raymond.cmpp.client.CmppClient;
import com.raymond.cmpp.client.msg.MessageManages;
import com.raymond.cmpp.common.thread.BaseCirculationThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 处理异步消息
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-08-08 11:39
 */
public class AsyncTaskHandleThread extends BaseCirculationThread {

    private final static Logger logger = LoggerFactory.getLogger(AsyncTaskHandleThread.class);

    private final BlockingQueue<MessageManages> asyncTask;

    private final ConcurrentHashMap<Integer, MessageManages> msgManager;

    public AsyncTaskHandleThread(BlockingQueue<MessageManages> asyncTask, CmppClient queueClient) {
        this.asyncTask = asyncTask;
        this.msgManager = queueClient.getMsgManager();
    }


    @Override
    protected void execute() {
        try {
            MessageManages take = asyncTask.take();
            if (take.isDone()) {
                return;
            }
//            long sleepTime = (take.getStartTime() + this.timeout) - CachedMillisecondClock.INS.now();
//            if (sleepTime > 0) {
//                System.out.println("异步任务剩余数据：" + asyncTask.size() + ",等待时间:" + sleepTime);
//            }
            Thread.sleep(take.getTimeout());
            if (!take.isAsyncDone()) {
                take.asyncSendFail();
                this.msgManager.remove(take.getReq().getSequenceId());
            }
        } catch (Exception e) {
            logger.error("处理异步线程异常:", e);
        }

    }


}
